package com.sali.service.data_processor;

import cn.hutool.json.JSONUtil;
import com.sali.entity.DataProcessTime;
import com.sali.entity.TransferObject;
import org.apache.commons.collections.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BlockingQueue;

public class ProcessThread extends Thread {

    public static volatile boolean running = true;

    // 数据处理器，执行具体的数据处理流程
    private DataProcessor dataProcessor;

    private BlockingQueue<List<TransferObject>> dataChannel;

    public ProcessThread(DataProcessor dataProcessor, BlockingQueue<List<TransferObject>> dataChannel) {
        this.dataProcessor = dataProcessor;
        this.dataChannel = dataChannel;
    }

    @Override
    public void run() {
        while (running) {
            try {
                Thread.sleep(5 * 1000);
            } catch (Exception ignore) {}

            // 从内存管道定时拉取数据
            List<TransferObject> transferObjectList = dataChannel.poll();
            if (CollectionUtils.isNotEmpty(transferObjectList)) {
                for (TransferObject transferObject : transferObjectList) {
                    if (transferObject == null || transferObject.getData() == null) {
                        continue;
                    }
                    transferObject.setProtocolCode(transferObject.getTopic());
                    transferObject.setProcessTime(DataProcessTime.getInstance());
                    this.dataProcessor.submit(transferObject);
                }
            }
        }
    }

}
